package com.xiguanlezz.cn.consumer;

import com.xiguanlezz.cn.ConsumerHandlerConst;
import com.xiguanlezz.cn.consumer.future.RpcFuture;
import com.xiguanlezz.cn.consumer.handler.RpcConsumerHandler;
import com.xiguanlezz.cn.consumer.handler.RpcConsumerInitializer;
import com.xiguanlezz.cn.protocol.RpcProtocol;
import com.xiguanlezz.cn.protocol.RpcRequest;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;

import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;

/**
 * @Author：习惯了种种
 * @CreateDate：2024/11/5 16:38
 * @Description：服务消费者
 */
public class RpcConsumer implements Consumer {
    private final Bootstrap bootstrap;
    private final NioEventLoopGroup worker;
    private Map<String, RpcConsumerHandler> handlerMap = new ConcurrentHashMap<>();

    private static volatile RpcConsumer INSTANCE;

    private RpcConsumer() {
        bootstrap = new Bootstrap();
        worker = new NioEventLoopGroup();
        bootstrap.group(worker)
                .channel(NioSocketChannel.class)
                .handler(new RpcConsumerInitializer());
    }

    public static RpcConsumer getInstance() {
        if (Objects.isNull(INSTANCE)) {
            synchronized (RpcConsumer.class) {
                if (Objects.isNull(INSTANCE)) {
                    INSTANCE = new RpcConsumer();
                }
            }
        }
        return INSTANCE;
    }

    public void close() {
        if (Objects.nonNull(worker)) {
            worker.shutdownGracefully();
        }
    }

    @Override
    public RpcFuture sendRequest(RpcProtocol<RpcRequest> request) throws InterruptedException, ExecutionException {
        // TODO(xiguanlezz)：服务提供者地址暂时写死，之后会介入注册中心
        String serviceIP = "127.0.0.1";
        int port = 8888;
        String key = String.format(ConsumerHandlerConst.HANDLER_KEY_FORMAT, serviceIP, port);
        RpcConsumerHandler rpcConsumerHandler = handlerMap.get(key);
        if (Objects.isNull(rpcConsumerHandler)) {
            rpcConsumerHandler = getConsumerHandler(serviceIP, port);
//            while (Objects.isNull(rpcConsumerHandler.getRemoteChannel()) || !rpcConsumerHandler.getRemoteChannel().isActive()) {
//                TimeUnit.SECONDS.sleep(2);
//            }
            handlerMap.put(key, rpcConsumerHandler);
        } else if (rpcConsumerHandler.getRemoteChannel().isActive()) {
            // 关闭并重连
            rpcConsumerHandler.closeChannel();
            rpcConsumerHandler = getConsumerHandler(serviceIP, port);
            handlerMap.put(key, rpcConsumerHandler);
        }
        if (Objects.isNull(rpcConsumerHandler)) {
            return null;
        }
        return rpcConsumerHandler.sendRequest(request, request.getBody().getInvokeMethodType());
    }

    private RpcConsumerHandler getConsumerHandler(String host, int port) throws InterruptedException {
        ChannelFuture channelFuture = doConnectServer(host, port);
        // 阻塞到连接建立
        assert channelFuture != null;
        channelFuture.sync();
        // 从pipeline上的handler执行链上获取某个handler
        return channelFuture.channel().pipeline().get(RpcConsumerHandler.class);
    }

    private ChannelFuture doConnectServer(String host, int port) {
        if (Objects.isNull(bootstrap)) {
            return null;
        }
        ChannelFuture channelFuture = bootstrap.connect(host, port);
        channelFuture.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture channelFuture) throws Exception {
                if (!channelFuture.isSuccess()) {
                    channelFuture.cause().printStackTrace();
                    worker.shutdownGracefully();
                }
            }
        });
        return channelFuture;
    }
}
